﻿#include "RpcSocketChannel.h"
#include "XyNet/Channel/Connection/RpcCryptoConnection.h"
#include "XyNet/Common/CrossObjectHolder.h"
#include "XyNet/Common/Exception.h"
#include "XyNet/Common/SetMapHelper.h"
#include "XyNet/Endpoint/RpcEndpoint.h"
#include "XyNet/Protocol/cpp/Rpc.service.h"
#include "XyNet/Service/RpcArgument.h"
#include "XyNet/Service/RpcCallback.h"
#include "XyNet/Service/RpcController.h"
#include "XyNet/Service/RpcMethod.h"
#include "XyNet/Service/RpcService.h"
#include "XyNet/Service/RpcServiceManager.h"

namespace XyNet
{
	IdGenerator RpcSocketChannel::_socketChannelIdGenerator;

	RpcSocketChannel::RpcSocketChannel(RpcEndpoint* endpoint,
		shared_ptr<boost::asio::ip::tcp::socket> socket)
		: _id(_socketChannelIdGenerator())
		, _endpoint(endpoint), _socket(socket)
		, _serviceManager(make_shared<RpcServiceManager>())
		, _totalHandleMessage(0),_totalDelay(0), _minDelay(0), _maxDelay(0) {}

	RpcSocketChannel::~RpcSocketChannel()
	{
		stop();
	}

	const Id& RpcSocketChannel::getId() const
	{
		return _id;
	}

	void RpcSocketChannel::clearUserDataAndService()
	{
		_userData.clear();
		_serviceManager->clear();
	}

	RpcEndpoint* RpcSocketChannel::getEndpoint()
	{
		return _endpoint;
	}

	shared_ptr<boost::asio::ip::tcp::socket> RpcSocketChannel::getSocket()
	{
		return _socket;
	}

	shared_ptr<RpcServiceManager> RpcSocketChannel::getServiceManager()
	{
		return _serviceManager;
	}

	void RpcSocketChannel::start()
	{
		getConnection()->start();
	}

	void RpcSocketChannel::stop()
	{
		_userData.clear();
		_socket->close();
		_connection.reset();
		_serviceManager->clear();

		_messageIdGenerator.reset();
		_idCallbackMapObject.createUniqueLock()->getObject()->clear();
		_cacheDatasObject.createUniqueLock()->getObject()->clear();

		_totalHandleMessage = 0;
		_totalDelay = chrono::steady_clock::duration(0);
		_minDelay = chrono::steady_clock::duration(0);
		_maxDelay = chrono::steady_clock::duration(0);
	}

	bool RpcSocketChannel::isConnectComplete() const
	{
		return _socket->is_open() && _connection && _connection->isConnectComplete();
	}

	void RpcSocketChannel::handleSocketException(const boost::system::error_code& error)
	{
		_endpoint->handleSocketException(_socket, error, shared_from_this());
	}

	void RpcSocketChannel::handleSocketChannelException(int errorCode)
	{
		_endpoint->handleSocketChannelException(shared_from_this(), errorCode);
	}

	void RpcSocketChannel::handleConnectionData(const Bytes& data)
	{
		if (_endpoint->getConfig()->getAutoHandleMessage())
			read(data);	// 会在网络线程执行
		else
			addCacheData(data);
	}

	bool RpcSocketChannel::hasCacheDatas() const
	{
		return !_cacheDatasObject.createSharedLock()->getObject()->empty();
	}

	void RpcSocketChannel::handleCacheDatas()
	{
		auto lock = _cacheDatasObject.createUpgradeLock();
		auto* cacheDatas = lock->getObject();
		if (!cacheDatas->empty())
		{
			vector<Bytes> datas;
			lock->toUnique()->swap(datas);
			lock->unlock();

			for (const Bytes& data : datas)
				read(data);
		}
	}

	vector<Bytes> RpcSocketChannel::popCacheDatas()
	{
		vector<Bytes> datas;
		auto lock = _cacheDatasObject.createUpgradeLock();
		auto* cacheDatas = lock->getObject();
		if (!cacheDatas->empty())
			lock->toUnique()->swap(datas);
		lock->unlock();
		return datas;
	}

	void RpcSocketChannel::setFailedCallback(const Id& messageId, int errorCode)
	{
		RpcMessage message;
		RpcMessage::Response* response = message.mutable_response();
		response->set_error_code(errorCode);
		handleConnectionData(message.SerializeAsString());
	}

	int RpcSocketChannel::getCallbackCount() const
	{
		return numeric_cast<int>(_idCallbackMapObject.createSharedLock()->getObject()->size());
	}

	chrono::steady_clock::duration RpcSocketChannel::getAverageDelay() const
	{
		if (_totalHandleMessage > 0)
			return _totalDelay / _totalHandleMessage;
		else
			return chrono::steady_clock::duration(0);
	}

	int RpcSocketChannel::getMaxReadMessageCountPerSecond() const
	{
		return _connection->getMaxReadMessageCountPerSecond();
	}

	int RpcSocketChannel::getMaxWriteMessageCountPerSecond() const
	{
		return _connection->getMaxWriteMessageCountPerSecond();
	}

	int RpcSocketChannel::getMaxReadBytesPerSecond() const
	{
		return _connection->getMaxReadBytesPerSecond();
	}

	int RpcSocketChannel::getMaxWriteBytesPerSecond() const
	{
		return _connection->getMaxWriteBytesPerSecond();
	}

	void RpcSocketChannel::callMethod(RpcMethod* method, shared_ptr<RpcCallback> callback)
	{
		// 存储回调
		Id messageId = _messageIdGenerator();
		SetMapHelper::insert(_idCallbackMapObject.createUniqueLock()->getObject(), 
			messageId, callback);

		// 根据连接状态决定是发消息还是返回错误
		if (!getSocket()->is_open())
		{
			setFailedCallback(messageId, RpcErrors::NotConnected);
		}
		else
		{
			// 设置定时器
			callback->startTimer(this, messageId);

			// 生成rpc消息
			RpcMessage message;
			message.set_id(messageId);
			auto request = message.mutable_request();
			request->set_data(callback->getArgument()->getRequest()->SerializeAsString());
			request->set_service_id(method->getServiceId());
			request->set_method_id(method->getId());

			// 发送消息
			write(message.SerializeAsString());
		}
	}

	//void RpcSocketChannel::sendData(const Bytes& data)
	//{
	//	write(data);
	//}

	void RpcSocketChannel::read(const Bytes& data)
	{
		parseMessage(data);
	}

	void RpcSocketChannel::write(const Bytes& data)
	{
		getConnection()->write(data);
	}

	void RpcSocketChannel::addCacheData(const Bytes& data)
	{
		_cacheDatasObject.createUniqueLock()->getObject()->emplace_back(data);
	}

	void RpcSocketChannel::parseMessage(const Bytes& data)
	{
		RpcMessage message;
		if (!message.ParseFromString(data))
		{
			handleSocketChannelException(RpcErrors::ParseMessageFailed);
		}
		else if (message.has_request())	// 处理请求
		{
			const RpcMessage::Request& request = message.request();
			// 先查找自带的服务
			auto service = _serviceManager->getService(request.service_id());
			if (!service)	// 再查找终端的服务
			{
				service = _endpoint->getServiceManager()->getService(
					request.service_id());
			}

			shared_ptr<RpcArgument> argument;
			shared_ptr<RpcController> controller;
			if (!service)
			{
				controller = make_shared<RpcController>(this);
				controller->setFailed(RpcErrors::ServiceNotFound);
			}
			else
			{
				argument = service->createMethodArgument(request.method_id());
				argument->initializeController(this);
				controller = argument->getController();

				if (!const_pointer_cast<google::protobuf::Message>(
					argument->getRequest())->ParseFromString(request.data()))
				{
					controller->setFailed(RpcErrors::ParseMessageFailed,
						"无法解析的消息（可能是版本低引起的，请更新到最新版）");
				}
				else
				{
					try
					{
						service->onMethod(request.method_id(), argument.get());
					}
					catch (const exception& e)
					{
						controller->setFailed(RpcErrors::LogicError, e.what());
					}
				}
			}

			// 回消息（消息id为0表示广播消息，不回）
			if (message.id() != 0)
			{
				message.clear_request();	// 清除请求，减少数据量
				RpcMessage::Response* response = message.mutable_response();
				if (controller->isFailed())
				{
					const auto& error = controller->getError();
					response->set_error_code(error->getCode());
					response->set_error_detail(error->getDetail());
				}
				else
				{
					response->set_data(argument->getResponse()->SerializeAsString());
				}

				write(message.SerializeAsString());
			}
		}
		else if (message.has_response())	// 处理应答
		{
			Id messageId = message.id();
			auto callback = popCallback(messageId);
			if (callback)
			{
				auto argument = callback->getArgument();
				argument->initializeController(this);
				auto controller = argument->getController();

				const RpcMessage::Response& response = message.response();
				if (response.error_code() != 0)
				{
					controller->setFailed(response.error_code(), response.error_detail());
				}
				else
				{
					if (!argument->getResponse()->ParseFromString(response.data()))
					{
						controller->setFailed(RpcErrors::ParseMessageFailed,
							"无法解析的消息（可能是版本低引起的，请更新到最新版）");
					}
				}

				callback->run();
			}
		}
		else
		{
			handleSocketChannelException(RpcErrors::ParseMessageFailed);
		}
	}

	shared_ptr<RpcCallback> RpcSocketChannel::popCallback(const Id& messageId)
	{
		shared_ptr<RpcCallback> callback;
		auto lock = _idCallbackMapObject.createUpgradeLock();
		auto* idCallbackMap = lock->getObject();
		auto iterator = idCallbackMap->find(messageId);
		if (iterator == idCallbackMap->end())
		{
			lock->unlock();
			//assert(false && "无法找到对应的回调");	// 有可能是正常处理和超时处理冲突
			return nullptr;
		}
		else
		{
			auto callback = iterator->second;
			lock->toUnique()->erase(iterator);
			lock->unlock();

			// 统计延时
			auto delay = chrono::steady_clock::now() - callback->getCreateTime();
			if (_totalHandleMessage == 0)
			{
				_minDelay = delay;
				_maxDelay = delay;
			}
			else
			{
				if (delay < _minDelay)
					_minDelay = delay;
				if (delay > _maxDelay)
					_maxDelay = delay;
			}
			++_totalHandleMessage;
			_totalDelay += delay;

			return callback;
		}
	}

	shared_ptr<RpcConnection> RpcSocketChannel::getConnection()
	{
		if (!_connection)
			_connection = make_shared<RpcConnection>(this);
		return _connection;
	}

}

//////////////////////////////////////////////////////////////////////////

namespace XyNet
{
	XYNET_API bool RpcSocketChannelHasCacheDatas(RpcSocketChannel* socketChannel)
	{
		try
		{
			return socketChannel->hasCacheDatas();
		}
		catch (const exception& e)
		{
			HandleException(e);
			return false;
		}
	}

	XYNET_API vector<Bytes>* RpcSocketChannelPopCacheDatas(
		RpcSocketChannel* socketChannel)
	{
		try
		{
			return BytesVectorHold(socketChannel->popCacheDatas());
		}
		catch (const exception& e)
		{
			HandleException(e);
			return nullptr;
		}
	}

	XYNET_API void RpcSocketChannelWriteData(
		RpcSocketChannel* socketChannel, string* data)
	{
		try
		{
			socketChannel->write(*data);
		}
		catch (const exception& e)
		{
			HandleException(e);
		}
	}
}